package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"log"
	"sync"
	"time"
)

func ProduceMsg(host string, config *sarama.Config) {
	msg := &sarama.ProducerMessage{}
	msg.Topic = "test_topic"
	msg.Value = sarama.StringEncoder("first message" + time.Now().String())
	client, err := sarama.NewSyncProducer([]string{host}, config)
	if err != nil {
		log.Fatal(err)
		return
	}
	defer client.Close()

	partition, offset, err := client.SendMessage(msg)
	if err != nil {
		log.Fatal("send msg failed: ", err)
		return
	}
	log.Printf("pid: %v offset: %v\n", partition, offset)
}

func ConsumerMsg(host string, config *sarama.Config) {
	consumer, err := sarama.NewConsumer([]string{host}, config)
	if err != nil {
		log.Println(err)
		return
	}

	partitionList, err := consumer.Partitions("test_topic")
	if err != nil {
		log.Println(err)
		return
	}

	var wg sync.WaitGroup
	wg.Add(1)
	for partition := range partitionList {
		//消费者 消费 对应主题的 具体 分区 指定 主题 分区 offset  return 对应分区的对象
		pc, err := consumer.ConsumePartition("test_topic", int32(partition), sarama.OffsetOldest)
		if err != nil {
			log.Println(err)
			return
		}

		// 运行完毕记得关闭
		defer pc.AsyncClose()

		// 去出对应的 消息
		// 通过异步 拿到 消息
		go func(sarama.PartitionConsumer) {
			defer wg.Done()
			for msg := range pc.Messages() {
				fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v\n",
					msg.Partition, msg.Offset, msg.Key, string(msg.Value))
			}
		}(pc)
	}
	wg.Wait()
}

func main() {
	Host := "192.168.133.2:9092"
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
	config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回
	//ProduceMsg(Host, config)
	config.Consumer.Group.Member.UserData = []byte("group1")
	config.Consumer.Offsets.AutoCommit.Enable = true
	ConsumerMsg(Host, config)
}
